package logic

import (
	"context"
	"log"
	"time"

	"go-exp/exp-test/kafka-test/kafka_client/internal/svc"
	"go-exp/exp-test/kafka-test/kafka_client/internal/types"

	kafka "github.com/segmentio/kafka-go"
	"github.com/zeromicro/go-zero/core/logx"
)

type Kafka_clientLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewKafka_clientLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Kafka_clientLogic {
	return &Kafka_clientLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *Kafka_clientLogic) Kafka_client(req *types.Request) (resp *types.Response, err error) {
	// todo: add your logic here and delete this line
	topic := "my-topic"
	partition := 0

	conn, err := kafka.DialLeader(l.ctx, "tcp", "192.168.168.190:29092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
	_, err = conn.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	if err := conn.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
	return
}
